import asyncio
import random
import sys
from datetime import date, datetime, timedelta
from typing import Iterable, List, Tuple

import akshare as ak
from sqlalchemy.ext.asyncio import AsyncSession

# 复用项目内部异步会话与DAO
from pathlib import Path
# 将项目根目录加入 sys.path，确保可导入 config 与 module_stock 包
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))
# 规避 config.env 的 argparse 仅允许 --env 的限制：先清空除脚本名以外的argv
sys.argv = [sys.argv[0]]

from config.database import AsyncSessionLocal
from module_stock.dao.basic_dao import BasicDao
from module_stock.dao.daily_dao import DailyDao


def _to_ak_date(d: date) -> str:
    return d.strftime("%Y%m%d")


def _safe_float(x):
    try:
        return None if x is None else float(x)
    except Exception:
        return None


def _convert_df_to_records(symbol: str, df) -> List[dict]:
    """
    将 AkShare 返回的 DataFrame 转换为 stock_daily 所需字段
    预期列名（ak.stock_zh_a_hist, adjust='qfq' 或 'hfq' 时）：
      日期, 开盘, 收盘, 最高, 最低, 成交量, 成交额, 振幅, 涨跌幅, 涨跌额, 换手率
    """
    records: List[dict] = []
    if df is None or df.empty:
        return records
    for _, r in df.iterrows():
        # 某些字段可能不存在或为None，尽量容错
        records.append(
            {
                "symbol": symbol,
                "trade_date": datetime.strptime(str(r.get("日期")), "%Y-%m-%d").date()
                if isinstance(r.get("日期"), str)
                else r.get("日期"),
                "open": _safe_float(r.get("开盘")),
                "close": _safe_float(r.get("收盘")),
                "high": _safe_float(r.get("最高")),
                "low": _safe_float(r.get("最低")),
                "vol": _safe_float(r.get("成交量")),
                "amount": _safe_float(r.get("成交额")),
                "amplitude": _safe_float(r.get("振幅")),
                "pct_chg": _safe_float(r.get("涨跌幅")),
                "chg": _safe_float(r.get("涨跌额")),
                "turn": _safe_float(r.get("换手率")),
            }
        )
    return records


async def _get_all_symbols(session: AsyncSession, limit: int | None) -> List[str]:
    symbols = await BasicDao.get_all_symbols(session)
    # BasicDao.get_all_symbols 返回的是列表（标量），直接切片
    if limit is not None and limit > 0:
        return symbols[:limit]
    return symbols


async def _sync_symbol_last_30_days(session: AsyncSession, symbol: str, end_day: date) -> Tuple[str, int]:
    """
    增量同步某只股票最近30天数据：
    - 起始日 = max(已存在最后一条K线日期+1, end_day-30)
    - 终止日 = end_day
    - 若无缺口，跳过
    """
    # 设定起始边界为最近30天
    boundary_start = end_day - timedelta(days=30)
    last = await DailyDao.get_last_daily_detail(session, symbol)
    if last and last.trade_date and last.trade_date >= boundary_start:
        start_day = last.trade_date + timedelta(days=1)
    else:
        start_day = boundary_start
    if start_day > end_day:
        return symbol, 0

    start_str = _to_ak_date(start_day)
    end_str = _to_ak_date(end_day)
    # 使用前复权，避免前端展示断裂
    df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_str, end_date=end_str, adjust="qfq")
    records = _convert_df_to_records(symbol, df)
    if not records:
        return symbol, 0
    await DailyDao.batch_add_daily_dao(session, records)
    return symbol, len(records)


async def main(limit: int | None = None, sleep_min: float = 0.8, sleep_max: float = 1.6):
    """
    同步所有（或前N只）股票最近30天日线，带随机限速，降低被封IP风险
    """
    async with AsyncSessionLocal() as session:
        end_day = date.today()
        symbols = await _get_all_symbols(session, limit)
        total_inserted = 0
        processed = 0
        for symbol in symbols:
            try:
                sym, cnt = await _sync_symbol_last_30_days(session, symbol, end_day)
                await session.commit()
                total_inserted += cnt
            except Exception as e:
                await session.rollback()
                # 打印并继续，不要中断批处理
                print(f"[WARN] {symbol} 同步失败: {e}", file=sys.stderr)
            finally:
                processed += 1
                # 随机sleep，降低被风控概率
                await asyncio.sleep(random.uniform(sleep_min, sleep_max))
                if processed % 50 == 0:
                    print(f"[INFO] 已处理 {processed}/{len(symbols)}，累计新增 {total_inserted}")
        print(f"[DONE] 完成 {processed} 只股票，同步新增 {total_inserted} 条")


if __name__ == "__main__":
    # 简单解析命令行参数：--limit N  --sleep-min  --sleep-max
    args = sys.argv[1:]
    kv = {args[i].lstrip("-"): args[i + 1] for i in range(0, len(args) - 1, 2)} if len(args) >= 2 else {}
    limit = int(kv.get("limit")) if "limit" in kv else None
    sleep_min = float(kv.get("sleep-min")) if "sleep-min" in kv else 0.8
    sleep_max = float(kv.get("sleep-max")) if "sleep-max" in kv else 1.6
    asyncio.run(main(limit=limit, sleep_min=sleep_min, sleep_max=sleep_max))


